package com.bungarus.busi;

import com.bungarus.Busi;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.BungarusClientException;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageType;
import com.bungarus.offline.OfflineMessagePersist;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by tang on 2018/11/25.
 */
@ChannelHandler.Sharable
public class ProcessMessageHandler extends ChannelInboundHandlerAdapter {
    Logger logger = Logger.getLogger(ProcessMessageHandler.class);
    private final Busi busiServer;

    public ProcessMessageHandler(Busi busServer) {
        this.busiServer = busServer;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof MessageProtobuf.Message){
            MessageProtobuf.Message message = (MessageProtobuf.Message) msg;
            switch (message.getHeader().getType()) {
                case MessageType.HEARTBEAT_PING:
                    //send back pong
                    ctx.writeAndFlush(MessageBuilder.buildHeartbeatPongMessage(message.getHeader().getId(), message.getHeader().getId2()));
                    break;
                case MessageType.AUTH:
                case MessageType.LOGOUT: {
                    logger.info("++++++++++ Message received : \n" + message.toString());
                    //Maintain access session info
                    //1. update the user's session in session container
                    Map<String, Object> data = new HashMap<>(3);
                    data.put(SessionManagement.SESSION_CONTEXT, ctx);
                    data.put(SessionManagement.SESSION, message);
                    data.put(SessionManagement.BUSI_SERVER, this.busiServer);
                    SessionManagement.getInstance().doBusiness(data);
                    //2. update the user's status ('online' or 'offline') in status container and broadcast the status to everyone who is
                    // in the same group with him.
                    StatusManagement.getInstance().doBusiness(data);
                    break;
                }
                case MessageType.STATUS: {
                    logger.info("++++++++++ Message received : \n" + message.toString());
                    //Maintain user online status
                    //1. update the user's status in container
                    //2. send the user's latest status to everyone who is in the same group with him.
                    Map<String, Object> data = new HashMap<>(3);
                    data.put(SessionManagement.SESSION_CONTEXT, ctx);
                    data.put(SessionManagement.SESSION, message);
                    data.put(SessionManagement.BUSI_SERVER, this.busiServer);

                    StatusManagement.getInstance().doBusiness(data);
                    break;
                }
                case MessageType.IM: {
                    logger.info("++++++++++ Message received : \n" + message.toString());
                    //send the im message to the receipt
                    //1. find receipt in current cluster first.
                    //2. find receipt in other clusters second if the person cannot be found in step 1.
                    ReceiptHunter receiptHunter = ReceiptHunter.getInstance();
                    List<ReceiptHunter.HunterInfo> receipts = receiptHunter.locateAll(message, busiServer);
                    //3. mark the message as offline message if the person cannot be found in step 2.
                    //4. save the offline message.

                    // if sessions exist, try to send it, if failed, the message will be persisted as RetryFailedMessage
                    receipts.stream().forEach(receipt -> {
                        try {
                            receipt.busi2AccessClient.sendMessage(
                                    MessageBuilder.build2AccessMessage(Busi.CLUSTER_TAG_ACCESS, message, receipt.sessionInAccess),
                                    Busi.CLUSTER_TAG_ACCESS, ctx);
                        }
                        catch (BungarusClientException e) {
                            e.printStackTrace();
                        }
                    });
                    // if no sessions at all, persist the message and retry
                    if (receipts.size() <= 0) {
                        OfflineMessagePersist offlinePersist = OfflineMessagePersist.getInstance(
                                busiServer.getMongoTemplate4FailedRetry());
                        offlinePersist.doBusiness(message);
                        ctx.writeAndFlush(MessageBuilder.buildAckMessage(message.getHeader().getId(), message.getHeader().getId2()));
                    }
                    // TODO: session query between busi clusters
                    break;
                }
                case MessageType.IM_GROUP: {
                    logger.info("++++++++++ Message received : \n" + message.toString());
                    GroupMessageProcessing processing = GroupMessageProcessing.getInstance(this.busiServer);
                    processing.doBusiness(message);
                    ctx.writeAndFlush(MessageBuilder.buildAckMessage(message.getHeader().getId(), message.getHeader().getId2()));
                    break;
                }
                case MessageType.OFFLINE: {//access node sent message which the real user cannot ack to busi node.
                    logger.info("++++++++++ Message received : \n" + message.toString());
                    OfflineMessagePersist offlinePersist = OfflineMessagePersist.getInstance(
                            busiServer.getMongoTemplate4FailedRetry());
                    offlinePersist.doBusiness(message);
                    ctx.writeAndFlush(MessageBuilder.buildAckMessage(message.getHeader().getId(), message.getHeader().getId2()));
                    break;
                }
                case MessageType.BUSI_PROCESS:
                    break;
                default:
                    //TODO: send the other type message. it is not necessary to mark the message as offline if
                    // the message cannot be sent.
                    break;
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
    }
}
